package com.bw.yk09;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class FlinkDIMAsyncIO extends RichAsyncFunction<Order_info_stats,Order_info_stats> {
    private Connection conn = null;
    PreparedStatement ps1 = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181", "", "");
        ps1 = conn.prepareStatement("select line from dim_base_dic where id = ?");
        conn.setAutoCommit(true);
    }

    @Override
    public void close() throws Exception {
        ps1.close();
        conn.close();
    }
    @Override
    public void asyncInvoke(Order_info_stats order_info, ResultFuture<Order_info_stats> resultFuture) throws Exception {
        String cargo_type = order_info.getCargo_type();

        //查询dim_user_info
        ps1.setString(1,cargo_type+"");
        ResultSet rs1 = ps1.executeQuery();
        if(rs1.next()){
            String line = rs1.getString("line");
            JSONObject ob = JSON.parseObject(line);
            order_info.setCargo_type_name(ob.getString("name"));
        }


        //返回宽表
        resultFuture.complete(Collections.singleton(order_info));
    }
}
